package com.zwg.tingshu.common.service;

import com.zwg.tingshu.common.constant.RocketMqConstant;
import com.zwg.tingshu.common.mqtemplate.ExtRocketMQTemplate;
import org.apache.rocketmq.client.producer.SendCallback;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.messaging.Message;
import org.springframework.messaging.MessagingException;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.List;

@Component
public class RocketMqService {
    private static final Logger logger = LoggerFactory.getLogger(RocketMqService.class);

    @Resource(name = "extRocketMQTemplate")
    private ExtRocketMQTemplate rocketMQTemplate;

    /**
     * 发送单向消息,即只负责发送消息,不等待应答,不关心结果
     */
    public  void sendOneWay(String topic,String tag, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            rocketMQTemplate.sendOneWay(topic+":"+tag,message);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送带key的单向消息,即只负责发送消息,不等待应答,不关心结果
     */
    public  void sendOneWay(String topic,String tag, String key, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", key)
                    .build();
            rocketMQTemplate.sendOneWay(topic+":"+tag,message);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }


    /**
     * 发送同步消息,即消息发送者需要等待服务器应答并收到应答后消息发送才完成,这种可靠性同步的方式一般用在重要的通知
     */
    public  void sendSync(String topic,String tag, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            SendResult sendResult = rocketMQTemplate.syncSend(topic + ":" + tag, message);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送带key的同步消息,即消息发送者需要等待服务器应答并收到应答后消息发送才完成,这种可靠性同步的方式一般用在重要的通知
     */
    public  void sendSync(String topic,String tag, String key, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", key)
                    .build();
            SendResult sendResult = rocketMQTemplate.syncSend(topic + ":" + tag, message);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送异步消息,即消息发送者只管发消息，服务器响应由回调函数处理
     */
    public void sendAsync(String topic,String tag, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            rocketMQTemplate.asyncSend(topic + ":" + tag, message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                }
                @Override
                public void onException(Throwable throwable) {
                    throw new RuntimeException(RocketMqConstant.MQ_ERROR);
                }
            });
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送带key的异步消息,即消息发送者只管发消息，服务器响应由回调函数处理
     */
    public void sendAsync(String topic,String tag, String key, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", key)
                    .build();
            rocketMQTemplate.asyncSend(topic + ":" + tag, message, new SendCallback() {
                @Override
                public void onSuccess(SendResult sendResult) {
                }
                @Override
                public void onException(Throwable throwable) {
                    throw new RuntimeException(RocketMqConstant.MQ_ERROR);
                }
            });
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送单向有序消息,即只负责发送消息,不等待应答,不关心结果
     */
    public  void sendOneWayOrderly(String topic,String tag,String orderId, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            rocketMQTemplate.sendOneWayOrderly(topic + ":" + tag, message,orderId);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送带key的单向有序消息,即只负责发送消息,不等待应答,不关心结果
     */
    public  void sendOneWayOrderly(String topic,String tag, String key, String orderId, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", key)
                    .build();
            rocketMQTemplate.sendOneWayOrderly(topic + ":" + tag, message,orderId);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送同步有序消息,即消息发送者需要等待服务器应答并收到应答后消息发送才完成,这种可靠性同步的方式一般用在重要的通知
     */
    public  void syncSendOrderly(String topic,String tag,String orderId, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic + ":" + tag, message, orderId);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送带key的同步有序消息,即消息发送者需要等待服务器应答并收到应答后消息发送才完成,这种可靠性同步的方式一般用在重要的通知
     */
    public  void syncSendOrderly(String topic,String tag,String key, String orderId, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", key)
                    .build();
            SendResult sendResult = rocketMQTemplate.syncSendOrderly(topic + ":" + tag, message, orderId);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送异步有序消息,即消息发送者只管发消息，服务器响应由回调函数处理
     */
    public  void asyncSendOrderly(String topic,String tag,String orderId, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            rocketMQTemplate.asyncSendOrderly(topic + ":" + tag, message,orderId, new SendCallback(){

                @Override
                public void onSuccess(SendResult sendResult) {
                }

                @Override
                public void onException(Throwable throwable) {
                    throw new RuntimeException(RocketMqConstant.MQ_ERROR);
                }
            });
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送带key的异步有序消息,即消息发送者只管发消息，服务器响应由回调函数处理
     */
    public  void asyncSendOrderly(String topic,String tag, String key, String orderId, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", key)
                    .build();
            rocketMQTemplate.asyncSendOrderly(topic + ":" + tag, message,orderId, new SendCallback(){

                @Override
                public void onSuccess(SendResult sendResult) {
                }

                @Override
                public void onException(Throwable throwable) {
                    throw new RuntimeException(RocketMqConstant.MQ_ERROR);
                }
            });
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送延时消息,即消息发送者需要等待服务器应答并收到应答后消息发送才完成,这种可靠性同步的方式一般用在重要的通知
     */
    public  void syncSendDelayTimeSeconds(String topic,String tag,long delay, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            SendResult sendResult = rocketMQTemplate.syncSendDelayTimeSeconds(topic + ":" + tag, message,delay);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 发送带key的延时消息,即消息发送者需要等待服务器应答并收到应答后消息发送才完成,这种可靠性同步的方式一般用在重要的通知
     */
    public  void syncSendDelayTimeSeconds(String topic,String tag,String key, long delay, String msg){
        SendResult sendResult = null;
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", "key07")
                    .build();
            sendResult = rocketMQTemplate.syncSendDelayTimeSeconds(topic + ":" + tag, message,delay);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 事务消息
     */
    public  void sendMessageInTransaction(String topic,String tag, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .build();
            TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic+":"+tag, message, null);
        } catch (MessagingException e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 带key的事务消息
     */
    public  void sendMessageInTransaction(String topic,String tag, String key, String msg){
        try {
            Message<String> message = MessageBuilder.withPayload(msg)
                    .setHeader("KEYS", key)
                    .build();
            TransactionSendResult transactionSendResult = rocketMQTemplate.sendMessageInTransaction(topic+":"+tag, message, null);
        } catch (MessagingException e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }


    /**
     * 批量消息
     */
    public  void syncSendBatchMessage(String topic,String tag, List<String> msg){
        try {
            List<Message<String>> messages = new ArrayList<>();
            for (String tmsg : msg) {
                Message<String> message = MessageBuilder.withPayload(tmsg)
                        .build();
                messages.add(message);
            }
            SendResult sendResult = rocketMQTemplate.syncSend(topic + ":" + tag, messages);
        } catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }
    }

    /**
     * 带key的批量消息
     */
    public  void syncSendBatchMessage(String topic,String tag, String key, List<String> msg){
        try{
            List<Message<String>> messages = new ArrayList<>();
            for (String tmsg : msg) {
                Message<String> message = MessageBuilder.withPayload(tmsg)
                        .setHeader("KEYS",key).build();
                messages.add(message);
            }
            SendResult sendResult = rocketMQTemplate.syncSend(topic + ":" + tag, messages);
        }catch (Exception e) {
            throw new RuntimeException(RocketMqConstant.MQ_ERROR);
        }

    }
}